use std::convert::Infallible;
use std::net::{SocketAddr, TcpListener};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response, StatusCode};
use hyper_util::rt::tokio::TokioIo;
use http_body_util::Full;
use bytes::Bytes;
use pdata::internal::data::protogen::collector::logs::v1::ExportLogsServiceRequest;
use pdata::internal::data::protogen::collector::metrics::v1::ExportMetricsServiceRequest;
use pdata::internal::data::protogen::collector::trace::v1::ExportTraceServiceRequest;
use rust_collector::command::{Command, CommandBroadcaster};
use rust_collector::config::{Config, ExporterConfig, Exporters, Processors, QueueConfig, Receivers};
use rust_collector::exporter::otlp_http_exporter::OtlpHttpExporter;
use rust_collector::consumer::{ConsumeLogs, ConsumeMetrics, ConsumeTraces};
use tokio::net::TcpListener as TokioTcpListener;
use tokio::time::sleep;

fn pick_free_port() -> u16 {
    let l = TcpListener::bind("127.0.0.1:0").expect("bind temp");
    let port = l.local_addr().unwrap().port();
    drop(l);
    port
}

async fn start_test_sink_server() -> (SocketAddr, Arc<AtomicUsize>, Arc<AtomicUsize>, Arc<AtomicUsize>) {
    let traces = Arc::new(AtomicUsize::new(0));
    let metrics = Arc::new(AtomicUsize::new(0));
    let logs = Arc::new(AtomicUsize::new(0));

    let traces_c = traces.clone();
    let metrics_c = metrics.clone();
    let logs_c = logs.clone();

    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    let listener = TokioTcpListener::bind(addr).await.unwrap();

    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            let io = TokioIo::new(stream);
            let traces_c = traces_c.clone();
            let metrics_c = metrics_c.clone();
            let logs_c = logs_c.clone();
            tokio::spawn(async move {
                let svc = service_fn(move |req: Request<hyper::body::Incoming>| {
                    let traces_c = traces_c.clone();
                    let metrics_c = metrics_c.clone();
                    let logs_c = logs_c.clone();
                    async move {
                        let resp: Response<Full<Bytes>> = if req.method() != Method::POST {
                            Response::builder()
                                .status(StatusCode::METHOD_NOT_ALLOWED)
                                .body(Full::new(Bytes::from_static(b"method not allowed")))
                                .unwrap()
                        } else {
                            match req.uri().path() {
                                "/v1/traces" => {
                                    traces_c.fetch_add(1, Ordering::Relaxed);
                                    Response::builder().status(StatusCode::OK).body(Full::new(Bytes::new())).unwrap()
                                }
                                "/v1/metrics" => {
                                    metrics_c.fetch_add(1, Ordering::Relaxed);
                                    Response::builder().status(StatusCode::OK).body(Full::new(Bytes::new())).unwrap()
                                }
                                "/v1/logs" => {
                                    logs_c.fetch_add(1, Ordering::Relaxed);
                                    Response::builder().status(StatusCode::OK).body(Full::new(Bytes::new())).unwrap()
                                }
                                _ => Response::builder().status(StatusCode::NOT_FOUND).body(Full::new(Bytes::new())).unwrap(),
                            }
                        };
                        Ok::<Response<Full<Bytes>>, Infallible>(resp)
                    }
                });
                let _ = http1::Builder::new().serve_connection(io, svc).await;
            });
        }
    });

    sleep(Duration::from_millis(60)).await;
    (addr, traces, metrics, logs)
}

fn temp_dir() -> String {
    let mut dir = std::env::temp_dir();
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    dir.push(format!("http_exp_test_{}", nanos));
    dir.to_string_lossy().to_string()
}

fn make_cfg(endpoint: String) -> Config {
    Config {
        receivers: Receivers::default(),
        exporters: Exporters {
            debug: false,
            grpc: ExporterConfig::default(),
            http: ExporterConfig {
                endpoint,
                queue: QueueConfig { capacity_items: 16, capacity_bytes: 0, persistent: temp_dir() },
            },
        },
        processors: Processors::default(),
    }
}

#[tokio::test]
async fn http_exporter_trigger_replay_sends_all_types() {
    let (addr, traces, metrics, logs) = start_test_sink_server().await;
    let endpoint = format!("http://{}", addr);
    let cfg = make_cfg(endpoint);

    let exp = OtlpHttpExporter::new(&cfg).await.expect("exporter new");

    // subscribe to commands
    let broadcaster = Arc::new(CommandBroadcaster::new());
    exp.subscribe_to_commands(broadcaster.clone());

    // append one of each
    <OtlpHttpExporter as ConsumeTraces>::consume(&*exp, ExportTraceServiceRequest { resource_spans: Vec::new() }).await;
    <OtlpHttpExporter as ConsumeMetrics>::consume(&*exp, ExportMetricsServiceRequest { resource_metrics: Vec::new() }).await;
    <OtlpHttpExporter as ConsumeLogs>::consume(&*exp, ExportLogsServiceRequest { resource_logs: Vec::new() }).await;

    // trigger replay
    let _ = broadcaster.send_command(Command::LinkUp);

    // wait for sender/reader to process
    for _ in 0..30 {
        if traces.load(Ordering::Relaxed) >= 1 && metrics.load(Ordering::Relaxed) >= 1 && logs.load(Ordering::Relaxed) >= 1 {
            break;
        }
        sleep(Duration::from_millis(100)).await;
    }

    assert!(traces.load(Ordering::Relaxed) >= 1, "traces sent");
    assert!(metrics.load(Ordering::Relaxed) >= 1, "metrics sent");
    assert!(logs.load(Ordering::Relaxed) >= 1, "logs sent");
}

#[tokio::test]
async fn http_exporter_retry_on_5xx_then_success() {
    // server that fails first attempt on traces, then succeeds
    let fail_once = Arc::new(AtomicUsize::new(0));
    let fail_once_c = fail_once.clone();
    let port = pick_free_port();
    let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
    let listener = TokioTcpListener::bind(addr).await.unwrap();
    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            let io = TokioIo::new(stream);
            let fail_once_c = fail_once_c.clone();
            tokio::spawn(async move {
                let svc = service_fn(move |req: Request<hyper::body::Incoming>| {
                    let fail_once_c = fail_once_c.clone();
                    async move {
                        let resp: Response<Full<Bytes>> = if req.method() == Method::POST && req.uri().path() == "/v1/traces" {
                            let n = fail_once_c.fetch_add(1, Ordering::Relaxed);
                            if n == 0 {
                                Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Full::new(Bytes::new())).unwrap()
                            } else {
                                Response::builder().status(StatusCode::OK).body(Full::new(Bytes::new())).unwrap()
                            }
                        } else {
                            Response::builder().status(StatusCode::OK).body(Full::new(Bytes::new())).unwrap()
                        };
                        Ok::<Response<Full<Bytes>>, Infallible>(resp)
                    }
                });
                let _ = http1::Builder::new().serve_connection(io, svc).await;
            });
        }
    });

    let endpoint = format!("http://{}", addr);
    let cfg = make_cfg(endpoint);
    let exp = OtlpHttpExporter::new(&cfg).await.expect("exporter new");
    let broadcaster = Arc::new(CommandBroadcaster::new());
    exp.subscribe_to_commands(broadcaster.clone());

    <OtlpHttpExporter as ConsumeTraces>::consume(&*exp, ExportTraceServiceRequest { resource_spans: Vec::new() }).await;
    let _ = broadcaster.send_command(Command::LinkUp);

    // wait for at least 2 attempts (1 fail, 1 success)
    for _ in 0..30 {
        if fail_once.load(Ordering::Relaxed) >= 2 {
            break;
        }
        sleep(Duration::from_millis(100)).await;
    }
    assert!(fail_once.load(Ordering::Relaxed) >= 2);
}


